﻿#include "box_network.hh"
#include "box_channel.hh"
#include "fixed_mem_pool.hh"

#include "../../thirdparty/klogger/klogger/interface/logger.h"

#include "../config/box_config.hh"

#include "../detail/box_alloc.hh"
#include "../detail/lang_impl.hh"

#include "../repo/src/include/root/rpc_root.h"
#include "../repo/src/include/root/rpc_stub.h"

#include "../util/box_debug.hh"
#include "../util/object_pool.hh"
#include "../util/os_util.hh"
#include "../util/string_util.hh"

#include "loop/tcp_loop.hh"
#include "loop/udp/udp_loop.hh"

#include <cstring>
#include <unordered_map>

namespace kratos {
namespace service {

static std::shared_ptr<BoxChannel> NullChannel;

BoxNetwork::BoxNetwork()
    : main_queue(&net_to_main_queue, &main_to_net_queue),
      net_queue(&main_to_net_queue, &net_to_main_queue) {}

BoxNetwork::~BoxNetwork() { stop(); }

auto BoxNetwork::start() -> bool {
  on_config_change();
  // 启动网络线程
  start_worker();
  return true;
}

auto BoxNetwork::start(const std::string &type) -> bool {
  if (!start_loop(type)) {
    return false;
  }
  return start();
}

auto BoxNetwork::stop() -> bool {
  running_ = false;
  // 等待网络线程退出
  if (!thread_exit_ && thread_.joinable()) {
    thread_.join();
  }
  // 销毁网络循环
  if (loop_) {
    loop_->stop();
  }
  // 清理未完成的网络事件
  NetEventData event_data;
  while (true) {
    auto result = main_queue.try_read(event_data);
    if (!result) {
      break;
    }
    event_data.clear();
  }
  listener_name_map_.clear();
  return true;
}

auto BoxNetwork::start_worker() -> void {
  box_channel_recv_buffer_len_ = get_config().get_box_channel_recv_buffer_len();
  running_ = true;
  thread_ = std::thread([&]() {
    // 设置当前线程的退出标志
    thread_exit_ = false;
    while (running_) {
      // 1. 运行网络主循环
      // 2. 处理队列的网络事件
      if (loop_) {
        loop_->worker_update();
      }
      network_thread_processor();
    }
    // 清理网络线程
    worker_cleanup();
  });
}

auto BoxNetwork::worker_cleanup() -> void {
  // 清理所有未处理的网络事件
  NetEventData event_data;
  while (true) {
    auto result = net_queue.try_read(event_data);
    if (!result) {
      break;
    }
    event_data.clear();
  }
  // 线程退出标志
  thread_exit_ = true;
}

auto BoxNetwork::on_config_change() -> void {
  get_config().add_reload_listener(
      "network", [&](const std::string &, const config::BoxConfig &new_config) {
        const auto &old_config = get_config();
        if (old_config.get_box_channel_recv_buffer_len() !=
            new_config.get_box_channel_recv_buffer_len()) {
          box_channel_recv_buffer_len_ =
              new_config.get_box_channel_recv_buffer_len();
          std::string reason1 =
              "box_channel_recv_buffer_len = " +
              std::to_string(old_config.get_box_channel_recv_buffer_len());
          std::string reason2 =
              "box_channel_recv_buffer_len = " +
              std::to_string(new_config.get_box_channel_recv_buffer_len());
          write_log(lang::LangID::LANG_RELOAD_INFO, klogger::Logger::WARNING,
                    reason1.c_str(), reason2.c_str());
        }
      });
}

auto BoxNetwork::start_loop(const std::string &loop_type) -> bool {
  if (loop_) {
    return true;
  }
  // TODO 可以替换为不同类型的网络循环
  if (loop_type == "tcp") {
    loop_ = kratos::make_shared_pool_ptr<kratos::loop::TcpLoop>(this);
  } else if (loop_type == "udp") {
    loop_ = kratos::make_shared_pool_ptr<kratos::loop::UdpLoop>(this);
  } else {
    loop_ = kratos::make_shared_pool_ptr<kratos::loop::TcpLoop>(this);
  }
  if (!loop_) {
    return false;
  }
  if (!loop_->start()) {
    loop_->stop();
    loop_.reset();
    return false;
  }
  return true;
}

void BoxNetwork::network_thread_processor() {
  while (true) {
    NetEventData event_data;
    auto result = net_queue.try_read(event_data);
    if (!result) {
      break;
    }
    try {
      if (loop_) {
        loop_->do_worker_event(event_data);
      }
    } catch (std::exception &e) {
      write_log(lang::LangID::LANG_UNEXPECTED_EXCEPTION,
                klogger::Logger::FAILURE, "BoxNetwork",
                util::demangle(typeid(e).name()).c_str(), e.what());
    }
    event_data.clear();
  }
}

auto BoxNetwork::listen_at(const std::string name, const std::string &host,
                           int port) -> bool {

  auto ip = util::get_host_ip(host);
  if (!util::is_ip_address(ip) || (port == 0) || (name.empty())) {
    return false;
  }
  if (!start_loop("tcp")) {
    return false;
  }
  NetEventData event_data;
  std::string real_name = name;
  if (real_name.empty()) {
    real_name = "unknown";
  }
  event_data.event_id = NetEvent::listen_request;
  event_data.listen_request.name = box_malloc(real_name.size() + 1);
  box_assert(this, !event_data.listen_request.name);

  event_data.listen_request.host = box_malloc(host.size() + 1);
  box_assert(this, !event_data.listen_request.host);

  event_data.listen_request.port = port;
  memset(event_data.listen_request.name, 0, real_name.size() + 1);
  memcpy(event_data.listen_request.name, real_name.c_str(), real_name.size());
  memset(event_data.listen_request.host, 0, host.size() + 1);
  memcpy(event_data.listen_request.host, host.c_str(), host.size());
  // 发送到网络线程
  if (!main_queue.send(event_data)) {
    event_data.clear();
    return false;
  }
  return true;
}

auto BoxNetwork::connect_to(const std::string name, const std::string &host,
                            int port, int timeout) -> bool {

  auto ip = util::get_host_ip(host);
  if (!util::is_ip_address(ip) || (port == 0) || (name.empty())) {
    return false;
  }
  if (!start_loop("tcp")) {
    return false;
  }
  NetEventData event_data;
  event_data.event_id = NetEvent::connect_request;
  event_data.connect_request.host = box_malloc(host.size() + 1);
  box_assert(this, !event_data.connect_request.host);

  event_data.connect_request.name = box_malloc(name.size() + 1);
  box_assert(this, !event_data.connect_request.name);

  event_data.connect_request.port = port;
  event_data.connect_request.timeout = timeout;
  memset(event_data.connect_request.host, 0, host.size() + 1);
  memcpy(event_data.connect_request.host, host.c_str(), host.size());
  memset(event_data.connect_request.name, 0, name.size() + 1);
  memcpy(event_data.connect_request.name, name.c_str(), name.size());
  // 发送到网络线程
  if (!main_queue.send(event_data)) {
    event_data.clear();
    return false;
  }
  return true;
}

auto BoxNetwork::close_channel(std::uint64_t id) -> bool {
  return enqueue_close_request(id);
}

auto BoxNetwork::do_event_main(NetEventData &event_data) -> void {
  switch (event_data.event_id) {
  case NetEvent::listen_response: {
    auto id = event_data.listen_response.channel_id;
    if (event_data.listen_response.success) {
      auto ptr = make_shared_pool_ptr<BoxChannel>(
          id, this, event_data.listen_response.name);

      box_assert(this, !ptr);

      main_channel_map_[id] = ptr;
      on_listen(event_data.listen_response.name, true, ptr);
    } else {
      on_listen(event_data.listen_response.name, false, NullChannel);
    }
  } break;
  case NetEvent::accept_notify: {
    auto id = event_data.accept_notify.channel_id;
    auto ptr = make_shared_pool_ptr<BoxChannel>(id, this,
                                                event_data.accept_notify.name);

    box_assert(this, !id);
    box_assert(this, !ptr);

    main_channel_map_[id] = ptr;
    on_accept(ptr);
  } break;
  case NetEvent::close_notify: {
    auto id = event_data.close_notify.channel_id;
    box_assert(this, !id);

    auto it = main_channel_map_.find(id);
    if (it == main_channel_map_.end()) {
      break;
    }
    auto ptr = it->second;

    box_assert(this, !ptr);
    // 关闭事件处理
    on_close(ptr);
    // 设置关闭标记
    ptr->set_close_flag();
    // 更新主线程管道表
    main_channel_map_.erase(it);
  } break;
  case NetEvent::connect_response: {
    if (event_data.connect_response.success) {
      auto id = event_data.connect_response.channel_id;
      box_assert(this, !id);

      auto ptr = make_shared_pool_ptr<BoxChannel>(
          id, this, event_data.connect_response.name);

      box_assert(this, !ptr);

      // 更新主线程管道表
      main_channel_map_[id] = ptr;
      // 成功
      on_connect(event_data.connect_response.name, true, ptr);
    } else {
      // 失败
      on_connect(event_data.connect_response.name, false, NullChannel);
    }
  } break;
  case NetEvent::recv_data_notify: {
    auto id = event_data.recv_data_notify.channel_id;
    box_assert(this, !id);

    auto it = main_channel_map_.find(id);
    if (it == main_channel_map_.end()) {
      break;
    }
    auto ptr = it->second;

    box_assert(this, !ptr);

    auto bytes = ptr->write_buffer(event_data.recv_data_notify.data_ptr,
                                   event_data.recv_data_notify.length);
    if (bytes != event_data.recv_data_notify.length) {
      ptr->close();
    } else {
      on_data(ptr);
    }
  } break;
  default:
    break;
  }
}

auto BoxNetwork::update() -> void {
  NetEventData event_data;
  while (main_queue.try_read(event_data)) {
    try {
      do_event_main(event_data);
    } catch (std::exception &e) {
      write_log(lang::LangID::LANG_UNEXPECTED_EXCEPTION,
                klogger::Logger::FAILURE, "BoxNetwork",
                util::demangle(typeid(e).name()).c_str(), e.what());
    }
    event_data.clear();
  }
}

auto BoxNetwork::get_channel(std::uint64_t id) const noexcept
    -> const std::shared_ptr<BoxChannel> & {
  auto it = main_channel_map_.find(id);
  if (it == main_channel_map_.end()) {
    return NullChannel;
  }
  return it->second;
}

BoxNetwork::SPSCQueuePair &BoxNetwork::get_net_queue() noexcept {
  return net_queue;
}

BoxNetwork::SPSCQueuePair &BoxNetwork::get_main_queue() noexcept {
  return main_queue;
}

auto BoxNetwork::get_listener_name_map() noexcept -> ListenerNameMap & {
  return listener_name_map_;
}

int BoxNetwork::get_channel_recv_buffer_len() const noexcept {
  return box_channel_recv_buffer_len_;
}

int BoxNetwork::enqueue_send_request(std::uint64_t id, const char *data,
                                     int size) {

  box_assert(this, !id);
  box_assert(this, !data);
  box_assert(this, !size);

  NetEventData event;
  event.event_id = NetEvent::send_data_notify;
  event.send_data_notify.channel_id = id;
  event.send_data_notify.data_ptr = box_malloc(size);
  box_assert(this, !event.send_data_notify.data_ptr);

  event.send_data_notify.length = size;
  memcpy(event.send_data_notify.data_ptr, data, size);

  // 发送到网络线程
  if (!main_queue.send(event)) {
    event.clear();
    return 0;
  }
  return size;
}

bool BoxNetwork::enqueue_close_request(std::uint64_t id) {
  box_assert(this, !id);

  NetEventData event;
  event.event_id = NetEvent::close_request;
  event.close_request.channel_id = id;
  return main_queue.send(event);
}

} // namespace service
} // namespace kratos
